package com.iqiyi.pps.epg.core.message;

import com.iqiyi.kiwi.message.*;
import com.iqiyi.kiwi.message.config.DestinationConfig;
import com.iqiyi.kiwi.message.impl.DestinationImpl;
import com.iqiyi.kiwi.message.impl.MessageServiceImpl;
import com.iqiyi.kiwi.utils.Configuration;
import com.iqiyi.kiwi.utils.JaxbBinder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by lemuel on 2016/11/21.
 */
public class MQServiceImpl extends MessageServiceImpl {
    private static final Logger logger = LoggerFactory.getLogger(MQServiceImpl.class);
    public static final boolean ENABLE_MQ = Configuration.getBoolean("activemq.enable");
    private String configFile = "message.xml";
    private Map<String, MessageClient> clientMap = new HashMap<String, MessageClient>(5);
    private Map<String, Destination> destinationMap = new HashMap<String, Destination>(20);

    @Override
    public void init() {
        if (!ENABLE_MQ) {
            logger.info("Activemq disabled");
            return;
        }
        logger.info("Activemq enabled");
        this.configure(this.configFile);
    }

    public void configure(String configFile) {
        InputStream is = this.getClass().getClassLoader().getResourceAsStream(configFile);
        try {
            JaxbBinder jaxbBinder = new JaxbBinder(MessageConfigs.class);
            MessageConfigs configs = jaxbBinder.fromXML(is);

            //初始化客户端、连接池
            for (EpgClientConfig clientConfig : configs.getClientConfig()) {
                if (!clientConfig.isSkip()) {
                    MessageClientFactory messageClientFactory = (MessageClientFactory) Class.forName(clientConfig.getFactory()).newInstance();
                    MessageClient client = messageClientFactory.createMessageClient(clientConfig.getProperties());
                    client.setProperties(clientConfig.getProperties());
                    client.setName(clientConfig.getName());
                    client.initialize();
                    clientMap.put(client.getName(), client);
                }
            }

            //初始化每一个消息目的地(Queue或topic，及其生产者和消费者)
            for (DestinationConfig destinationConfig : configs.getDestinationConfigs()) {
                MessageClient client = this.clientMap.get(destinationConfig.getClient());
                if (client == null) {
                    continue;
                }
                DestinationImpl destination = new DestinationImpl(destinationConfig.getName(), destinationConfig.getType(), client);
                destination.setProcessor(destinationConfig.getProcessor());
                destination.setMode(destinationConfig.getMode());
                destination.setProperties(destinationConfig.getProperties());
                destination.setClient(client);
                destination.initialize();
                logger.info("init destination:"+destination.getName());
                destinationMap.put(destination.getName(), destination);
            }

        } catch (Exception e) {
            throw new RuntimeException("message configure error: ", e);
        }
    }

    /**
     * 发送消息
     *
     * @param destination 消息目的地
     * @param message     消息对象
     */
    @Override
    public void send(String destination, String message) {
        this.send(destination, message, DEFAULT_PRIORITY);
    }

    /**
     * 发送消息
     *
     * @param destination 消息目的地
     * @param message     消息对象
     * @param priority    优先级
     */
    @Override
    public void send(String destination, String message, int priority) {
        Destination dest = this.destinationMap.get(destination);
        if (dest != null) {
            Producer producer = dest.getProducer();
            producer.send(message, priority);

        }
    }

    /**
     * 发送MAP消息
     *
     * @param destination 消息目的地
     * @param message     消息对象
     */
    @Override
    public void send(String destination, Map<String, String> message) {
        this.send(destination, message, DEFAULT_PRIORITY);
    }

    /**
     * 发送MAP消息
     *
     * @param destination 消息目的地
     * @param message     消息对象
     * @param priority    优先级
     */
    @Override
    public void send(String destination, Map<String, String> message, int priority) {
        Destination dest = this.destinationMap.get(destination);
        if (dest != null) {
            Producer producer = dest.getProducer();
            producer.send(message, priority);
        }
    }


    /**
     * 发送消息
     * @param message     消息对象
     */
    @Override
    public void send(Message message) {
        Destination dest = this.destinationMap.get(message.getDestination());
        logger.info(" message.getDestination start: [{}][{}]",message.getDestination(),message.getMap());
        if (dest != null) {
            Producer producer = dest.getProducer();
            logger.info("message.getDestination  before send");
            boolean retValue =  producer.send(message);
            logger.info(" message.getDestination end: [{}][{}]",message.getMap(),retValue);
        }
    }


    /**
     * 接收消息
     *
     * @param destination 消息目的地
     * @param count       接收数量
     */
    @Override
    public List<String> receiveString(String destination, int count) {
        List<String> list = new ArrayList<String>(count);
        Destination dest = this.destinationMap.get(destination);

        if (dest != null && Destination.MODE_GET.equals(dest.getMode())) {
            Consumer consumer = dest.getConsumer();
            if (consumer != null) {
                for (int i = 0; i < count; i++) {
                    list.add(consumer.receiveString());
                }
            }
        }

        return list;
    }

    /**
     * 接收消息
     *
     * @param destination 消息目的地
     * @param count       接收数量
     */
    @Override
    public List<Map<String, String>> receiveMap(String destination, int count) {
        List<Map<String, String>> list = new ArrayList<Map<String, String>>(count);
        Destination dest = this.destinationMap.get(destination);

        if (dest != null && Destination.MODE_GET.equals(dest.getMode())) {
            Consumer consumer = dest.getConsumer();
            if (consumer != null) {
                for (int i = 0; i < count; i++) {
                    list.add(consumer.receiveMap());
                }
            }
        }
        return list;
    }
}
